package com.xyy.saas.payment.util.mq;

import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigChangeListener;
import com.ctrip.framework.apollo.ConfigService;
import com.ctrip.framework.apollo.model.ConfigChange;
import com.ctrip.framework.apollo.model.ConfigChangeEvent;
import com.google.common.util.concurrent.RateLimiter;
import com.xyy.saas.payment.adpater.pingxx.domain.PayDO;
import com.xyy.saas.payment.callback.core.result.CallbackRunner;
import com.xyy.saas.payment.common.Constants;
import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.cores.vo.CallbackResultVo;
import com.xyy.saas.payment.dao.model.PayResultOrder;
import com.xyy.saas.payment.service.PayResultOrderService;
import com.xyy.saas.payment.util.ApolloUtil;
import com.xyy.saas.payment.util.JSONUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashSet;
import java.util.Set;

@DependsOn("apolloUtil")
@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_WX_PAPPAY_CALLBACK, consumerGroup = RocketMqGroup.GROUP_WX_PAPPAY_CALLBACK, consumeMode = ConsumeMode.ORDERLY)
public class WxPapPayCallback implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener , InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(WxPapPayCallback.class);
    @Autowired
    private CallbackRunner callbackRunner;
    @Autowired
    private PayResultOrderService payResultOrderService;
    @Autowired
    private MqProducer mqProducer;

    private int rateLimiterCreate = Integer.parseInt(ApolloUtil.getProperty(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE));

    private RateLimiter rateLimiter = RateLimiter.create(rateLimiterCreate);

    @Override
    public void onMessage(MessageExt messageExt) {
        rateLimiter.acquire(1);
        String body = new String(messageExt.getBody());
        PayDO payDO = JSONUtils.parse2Object(body,  PayDO.class);
        if (payDO==null){
            LOGGER.error("WxPapPayCallback callBackHandler wxpappay callback fail domain is null");
            return;
        }
        try{
            PayResultOrder payResultOrder;
            payResultOrder = payResultOrderService.selectByPayNoAndChannel(payDO.getOrderNo(), Constants.CHANNEL_WX_NAME);
            if (payResultOrder!=null){
                LOGGER.error("WxPapPayCallback callBackHandler wxPapPay callback repeat payNo:{}", payDO.getOrderNo());
                return;
            }
            payResultOrder = new PayResultOrder();
            payResultOrder.setPayNo(payDO.getOrderNo());
            payResultOrder.setChannelChargeNo(payDO.getChannelChargeNo());
            payResultOrder.setPayChannel(Constants.CHANNEL_DIRECT_NAME_WX);
            payResultOrder.setBusinessType(payDO.getBusinessType());
            payResultOrder.setBusinessOrderType(payDO.getBusinessOrderType());
            payResultOrder.setAmount(payDO.getAmount());
            payResultOrder.setFee(0);
            payResultOrder.setReceiveId(payDO.getReceiverId());
            payResultOrder.setCtime(new Date());
            payResultOrder.setUtime(new Date());
            payResultOrder.setRefund((byte)0);
            payResultOrder.setRoyalty((byte)0);
            payResultOrder.setPayMode(payDO.getPayMode());
            payResultOrder.setMerchantId(payDO.getMchId());
            int count = payResultOrderService.insertSelective(payResultOrder);
            if (count<1){
                LOGGER.error("WxPapPayCallback callBackHandler wxpappay callback insert payResultOrder fail payResultOrder:{}", JSON.toJSON(payResultOrder));
                throw new Exception();
            }
            Long unixTime = System.currentTimeMillis()/1000;
            // 异步回调
            callbackRunner.callback(payResultOrder.getPayNo(), CallbackResultVo.TYPE_CHARGE_SUCCESS, payDO.getAmount(), payResultOrder.getFee(), unixTime, payDO.getCallbackAddress(), Constants.CHANNEL_WX_NAME);
            LOGGER.info("WxPapPayCallback callBackHandler wxpappay end payResultOrder:{}", JSON.toJSON(payResultOrder));
        }catch (Exception e){
            LOGGER.error("WxPapPayCallback callBackHandler error domain:{}", JSON.toJSON(payDO));
            if (payDO.getCount()==16){
                LOGGER.error("WxPapPayCallback callBackHandler error is 16 times domain:{}", JSON.toJSON(payDO));
                return ;
            }
            payDO.setCount(payDO.getCount()+1);
            mqProducer.syncSendDelayMsg(RocketMqTopic.TOPIC_WX_PAPPAY_CALLBACK, JSON.toJSONString(payDO), 1);
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        defaultMQPushConsumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Set set = new HashSet();
        set.add(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE);
        //动态监听配置变化
        Config config = ConfigService.getAppConfig();
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                ConfigChange change = changeEvent.getChange(Constants.WX_CARLLBACK_RATE_LIMITER_CREATE);
                rateLimiter = RateLimiter.create(Integer.parseInt(change.getNewValue()));
            }
        }, set);
    }
}
